package com.lagou.client;

import com.lagou.server.service.encoder.RpcEncoder;
import com.lagou.server.service.entity.RpcRequest;
import com.lagou.server.service.serializer.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

public class RpcClients {

    private static Map<RpcServer, UserClientHandler> userClientHandlers = new ConcurrentHashMap<>();

    static {
        InputStream resourceAsStream =
                RpcClients.class.getClassLoader()
                        .getResourceAsStream("rpcClient.properties");
        if (resourceAsStream == null) {
            throw new RuntimeException("客户端配置文件不存在");
        }
        Properties properties = new Properties();
        try {
            properties.load(resourceAsStream);
            String rpcServerRegisterBasePath = properties.getProperty("rpc.server-register-base-path");
            String rpcServerRegisterZkUrl = properties.getProperty("rpc.server-register-zk-url");
            int sessionTimeoutMs = 50000;
            int connectionTimeoutMs = 50000;
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);

            CuratorFramework client = CuratorFrameworkFactory
                    .builder().connectString(rpcServerRegisterZkUrl)
                    .sessionTimeoutMs(sessionTimeoutMs)
                    .connectionTimeoutMs(connectionTimeoutMs)
                    .retryPolicy(retryPolicy).build();

            client.start();

            List<String> serverPaths = client.getChildren().forPath(rpcServerRegisterBasePath);

            System.out.println(serverPaths);
            for (String serverPath : serverPaths) {
                byte[] bytes = client.getData().forPath(rpcServerRegisterBasePath + "/" + serverPath);
                String serverInfo = new String(bytes, "GBK");
                String[] ipWithPort = serverInfo.split(":");
                String ip = ipWithPort[0];
                int port = Integer.parseInt(ipWithPort[1]);
                RpcServer rpcServer = new RpcServer(ip, port);
                UserClientHandler userClientHandler = initClient(rpcServer);
                userClientHandlers.put(rpcServer, userClientHandler);
            }

            PathChildrenCache cache = new PathChildrenCache(client, rpcServerRegisterBasePath, true);
            cache.start();

            PathChildrenCacheListener cacheListener = (client1, event) -> {
                System.out.println("Event Type ：" + event.getType());
                switch (event.getType()) {
                    case CHILD_ADDED:
                        byte[] serverInfoBytes = event.getData().getData();
                        String serverInfo = new String(serverInfoBytes, "GBK");
                        String[] ipWithPort = serverInfo.split(":");
                        String ip = ipWithPort[0];
                        int port = Integer.parseInt(ipWithPort[1]);
                        RpcServer rpcServer = new RpcServer(ip, port);
                        UserClientHandler userClientHandler = initClient(rpcServer);
                        userClientHandlers.put(rpcServer, userClientHandler);
                        break;
                    case CHILD_REMOVED:
                        byte[] removedServerInfoBytes = event.getData().getData();
                        String serverInfo2 = new String(removedServerInfoBytes, "GBK");
                        System.out.println(serverInfo2);
                        String[] ipWithPort2 = serverInfo2.split(":");
                        String ip2 = ipWithPort2[0];
                        int port2 = Integer.parseInt(ipWithPort2[1]);
                        RpcServer rpcServer2 = new RpcServer(ip2, port2);
                        userClientHandlers.remove(rpcServer2);
                        break;
                    default:
                        break;
                }
            };
            cache.getListenable().addListener(cacheListener);

        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static UserClientHandler getUserClientHandler() {
        Set<RpcServer> rpcServers = userClientHandlers.keySet();
        if (rpcServers.size() == 0) return null;
        final long currentTime = System.currentTimeMillis();
        List<RpcServer> preProcRpcServers = rpcServers.stream().peek(rpcServer -> {
            if (currentTime - rpcServer.lastProcTime >= 5000) {
                // 最后一次调用时间失效
                rpcServer.setLastProcDuration(-1);
            }
        }).collect(Collectors.toList());
        Collections.shuffle(preProcRpcServers);
        preProcRpcServers.sort(new Comparator<RpcServer>() {
            @Override
            public int compare(RpcServer rpcServer1, RpcServer rpcServer2) {
                return rpcServer1.getLastProcDuration() - rpcServer2.getLastProcDuration();
            }
        });

        return userClientHandlers.get(preProcRpcServers.get(0));
    }

    //2.初始化netty客户端
    private static UserClientHandler initClient(RpcServer rpcServer) throws InterruptedException {
        UserClientHandler userClientHandler = new UserClientHandler();
        userClientHandler.setRpcServer(rpcServer);

        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(userClientHandler);
                    }
                });

        bootstrap.connect(rpcServer.getHostname(), rpcServer.getPort()).sync();
        return userClientHandler;
    }


    public static class RpcServer {
        private String hostname;
        private int port;
        // 服务端最后一次调用耗时
        private int lastProcDuration;
        // 服务端最后一次调用时间
        private long lastProcTime = System.currentTimeMillis();

        public RpcServer(String hostname, int port) {
            this.hostname = hostname;
            this.port = port;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            RpcServer rpcServer = (RpcServer) o;
            return port == rpcServer.port &&
                    Objects.equals(hostname, rpcServer.hostname);
        }

        @Override
        public int hashCode() {
            return Objects.hash(hostname, port);
        }

        public String getHostname() {
            return hostname;
        }

        public void setHostname(String hostname) {
            this.hostname = hostname;
        }

        public int getPort() {
            return port;
        }

        public void setPort(int port) {
            this.port = port;
        }

        public int getLastProcDuration() {
            return lastProcDuration;
        }

        public void setLastProcDuration(int lastProcDuration) {
            this.lastProcDuration = lastProcDuration;
        }

        public long getLastProcTime() {
            return lastProcTime;
        }

        public void setLastProcTime(long lastProcTime) {
            this.lastProcTime = lastProcTime;
        }

        @Override
        public String toString() {
            return "RpcServer{" +
                    "hostname='" + hostname + '\'' +
                    ", port=" + port +
                    ", lastProcDuration=" + lastProcDuration +
                    ", lastProcTime=" + lastProcTime +
                    '}';
        }
    }
}
